
package com.base.data.elasticsearch.core;

import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.gson.JsonObject;
import com.base.data.elasticsearch.annotations.Document;
import com.base.data.elasticsearch.annotations.Mapping;
import com.base.data.elasticsearch.annotations.Setting;
import com.base.data.elasticsearch.core.aggregation.AggregatedPage;
import com.base.data.elasticsearch.core.aggregation.impl.AggregatedPageImpl;
import com.base.data.elasticsearch.core.convert.BaseElasticsearchConverter;
import com.base.data.elasticsearch.core.convert.BaseMappingElasticsearchConverter;
import com.base.data.elasticsearch.core.listener.AsyncNotifyOnceListener;
import com.base.data.elasticsearch.core.mapping.BaseElasticsearchPersistentEntity;
import com.base.data.elasticsearch.core.mapping.BaseElasticsearchPersistentProperty;
import com.base.data.elasticsearch.core.mapping.BaseMappingBuilder;
import com.base.data.elasticsearch.core.mapping.BaseSimpleElasticsearchMappingContext;
import com.base.data.elasticsearch.core.operation.BaseElasticsearchOperations;
import com.base.data.elasticsearch.core.query.*;
import com.base.data.elasticsearch.core.query.simplify.SearchQuery;
import com.base.data.elasticsearch.core.syncmapping.BaseCreateIndexRequest;
import com.base.data.elasticsearch.core.syncmapping.RequestConverteUtil;
import com.base.data.elasticsearch.util.*;
import org.apache.http.ContentTooLongException;
import org.apache.http.Header;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.message.BasicHeader;
import org.apache.http.util.EntityUtils;
import org.apache.logging.log4j.util.Strings;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.collect.MapBuilder;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.util.CloseableIterator;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.stream.Collectors;

import static com.base.data.elasticsearch.core.mapping.BaseMappingBuilder.FIELD_PROPERTIES;
import static java.util.Collections.EMPTY_SET;
import static java.util.Collections.emptySet;
import static org.elasticsearch.index.VersionType.EXTERNAL;
import static org.elasticsearch.index.query.QueryBuilders.wrapperQuery;
import static org.springframework.util.ObjectUtils.isEmpty;

/**
 * ElasticsearchTemplate
 */
public class BaseElasticsearchTemplate implements BaseElasticsearchOperations, ApplicationContextAware {

	private static final Logger logger = LoggerFactory.getLogger(BaseElasticsearchTemplate.class);
	private BaseRestHighLevelClient client;
	private RestClient restClient;
	private BaseElasticsearchConverter elasticsearchConverter;
	private BaseResultsMapper resultsMapper;
	private String searchTimeout;
	private Boolean autoMapping = true;

	public BaseElasticsearchTemplate(BaseRestHighLevelClient client,RestClient restClient) {
		this(client,restClient, new BaseMappingElasticsearchConverter(new BaseSimpleElasticsearchMappingContext()));
	}

	public BaseElasticsearchTemplate(BaseRestHighLevelClient client,RestClient restClient, BaseEntityMapper entityMapper) {
		this(client,restClient, new BaseMappingElasticsearchConverter(new BaseSimpleElasticsearchMappingContext()), entityMapper);
	}

	public BaseElasticsearchTemplate(BaseRestHighLevelClient client,RestClient restClient, BaseElasticsearchConverter elasticsearchConverter,
                                     BaseEntityMapper entityMapper) {
		this(client,restClient, elasticsearchConverter,
				new BaseDefaultResultMapper(elasticsearchConverter.getMappingContext(), entityMapper));
	}

	public BaseElasticsearchTemplate(BaseRestHighLevelClient client,RestClient restClient, BaseResultsMapper resultsMapper) {
		this(client,restClient, new BaseMappingElasticsearchConverter(new BaseSimpleElasticsearchMappingContext()), resultsMapper);
	}

	public BaseElasticsearchTemplate(BaseRestHighLevelClient client,RestClient restClient, BaseElasticsearchConverter elasticsearchConverter) {
		this(client,restClient, elasticsearchConverter, new BaseDefaultResultMapper(elasticsearchConverter.getMappingContext()));
	}

	public BaseElasticsearchTemplate(BaseRestHighLevelClient client,
									 RestClient restClient,
									 BaseElasticsearchConverter elasticsearchConverter,
                                     BaseResultsMapper resultsMapper) {

		Assert.notNull(client, "Client must not be null!");
		Assert.notNull(restClient, "Client must not be null!");
		Assert.notNull(elasticsearchConverter, "ElasticsearchConverter must not be null!");
		Assert.notNull(resultsMapper, "ResultsMapper must not be null!");

		this.client = client;
		this.restClient = restClient;
		this.elasticsearchConverter = elasticsearchConverter;
		this.resultsMapper = resultsMapper;
	}

	public <T> T execute(ClientCallback<T> callback) {
		Assert.notNull(callback, "callback must not be null");
		try {
			return callback.doWithClient(client);
		} catch (IOException | RuntimeException exception) {
			if( exception instanceof ContentTooLongException) {
				throw new RuntimeException(((ContentTooLongException) exception).getMessage() + ",please fixed config : [elasticsearch.httpBufferLimit]");
			}
			RuntimeException runtimeException = exception instanceof RuntimeException ? (RuntimeException) exception
					: new RuntimeException(exception.getMessage(), exception);
			throw runtimeException;
		}
	}

	public void executeAsync(ClientCallbackAsync<?> callback) {
		Assert.notNull(callback, "callback must not be null");
		try {
			callback.doWithClient(client);
		} catch (IOException | RuntimeException exception) {
			if( exception instanceof ContentTooLongException) {
				throw new RuntimeException(((ContentTooLongException) exception).getMessage() + ",please fixed config : [elasticsearch.httpBufferLimit]");
			}
			RuntimeException runtimeException = exception instanceof RuntimeException ? (RuntimeException) exception
					: new RuntimeException(exception.getMessage(), exception);
			throw runtimeException;
		}
	}

	@Override
	public RestHighLevelClient getClient() {
		return client;
	}

	public void setSearchTimeout(String searchTimeout) {
		this.searchTimeout = searchTimeout;
	}

	public void setAutoMapping(Boolean autoMapping) {
		if(!autoMapping) {
			logger.warn("Elasticsearch Base conf AutoMapping is false!");
		}
		this.autoMapping = autoMapping;
	}

	@Override
	public <T> boolean createIndex(Class<T> clazz) {
		return createIndexIfNotCreated(clazz);
	}

	@Override
	public boolean createIndex(String indexName) {
		Assert.notNull(indexName, "No index defined for Query");
		boolean response = false;
		final CreateIndexRequest indexRequest = Requests.createIndexRequest(indexName);
		try{
			 response = client.performRequest(indexRequest, RequestConverteUtil::index, BaseRestHighLevelClient::convertCreateIndexResponse, emptySet());
		} catch (IOException e) {
			logger.error("createIndex",e);
		} catch (ElasticsearchStatusException e) {
			logger.error("createIndex",e);
		}
		return response;
	}

	@Override
	public <T> boolean putMapping(Class<T> clazz) {
		if (clazz.isAnnotationPresent(Mapping.class)) {
			String mappingPath = clazz.getAnnotation(Mapping.class).mappingPath();
			if (Strings.isNotEmpty(mappingPath)) {
				String mappings = readFileFromClasspath(mappingPath);
				if (Strings.isNotEmpty(mappings)) {
					return putMapping(clazz, mappings);
				}
			} else {
				logger.info("mappingPath in @Mapping has to be defined. Building mappings using @Field");
			}
		}
		BaseElasticsearchPersistentEntity<T> persistentEntity = getPersistentEntityFor(clazz);
		XContentBuilder xContentBuilder = null;
		try {

			BaseElasticsearchPersistentProperty property = persistentEntity.getRequiredIdProperty();

			xContentBuilder = BaseMappingBuilder.buildMapping(clazz, persistentEntity.getIndexType(),
					property.getFieldName(), persistentEntity.getParentType());
		} catch (Exception e) {
			throw new ElasticsearchException("Failed to build mapping for " + clazz.getSimpleName(), e);
		}
		return putMapping(clazz, xContentBuilder);
	}

	@Override
	public <T> boolean putMapping(Class<T> clazz, Object mapping) {
		return putMapping(getPersistentEntityFor(clazz).getIndexName(), getPersistentEntityFor(clazz).getIndexType(),
				mapping);
	}

	@Override
	public boolean putMapping(String indexName, String type, Object mapping) {
		Assert.notNull(indexName, "No index defined for putMapping()");
		Assert.notNull(type, "No type defined for putMapping()");
		boolean response = false;
		try{
			BaseCreateIndexRequest baseCreateIndexRequest = new BaseCreateIndexRequest();
			XContentBuilder builder = (XContentBuilder) mapping;
			baseCreateIndexRequest.index(indexName).type(type).create(false);
			if (mapping instanceof String) {
				baseCreateIndexRequest.source(String.valueOf(mapping));
			} else if (mapping instanceof Map) {
				baseCreateIndexRequest.source((Map) mapping);
			} else if (mapping instanceof XContentBuilder) {
				baseCreateIndexRequest.source((XContentBuilder) mapping);
			}
			logger.debug("index:[{}],type:[{}],create-mapping:[{}]",indexName,type,builder.string());
			response = client.performRequest(baseCreateIndexRequest, RequestConverteUtil::index, BaseRestHighLevelClient::convertCreateIndexResponse, emptySet());
			logger.debug("index:[{}],type:[{}],create-mapping:{}",indexName,type,response);
		} catch (IOException | ElasticsearchStatusException e) {
			logger.error("putIndexMapping",e);
			throw new com.base.data.elasticsearch.ElasticsearchException(e.getMessage());
		}
		return response;
	}

	@Override
	public Map getMapping(String indexName, String type) {
		Assert.notNull(indexName, "No index defined for putMapping()");
		Assert.notNull(type, "No type defined for putMapping()");
		Map mappings = null;
		try {
//			mappings = client.admin().indices().getMappings(new GetMappingsRequest().indices(indexName).types(type))
//					.actionGet().getMappings().get(indexName).get(type).getSourceAsMap();
			logger.error("404");
		} catch (Exception e) {
			throw new ElasticsearchException(
					"Error while getting mapping for indexName : " + indexName + " type : " + type + " " + e.getMessage());
		}
		return mappings;
	}

	@Override
	public <T> Map getMapping(Class<T> clazz) {
		return getMapping(getPersistentEntityFor(clazz).getIndexName(), getPersistentEntityFor(clazz).getIndexType());
	}

	@Override
	public Boolean autoMapping() {
		return this.autoMapping;
	}

	@Override
	public BaseElasticsearchConverter getElasticsearchConverter() {
		return elasticsearchConverter;
	}

	@Override
	public <T> T queryForObject(BaseGetQuery query, Class<T> clazz) {
		return queryForObject(query, clazz, resultsMapper);
	}

	@Override
	public <T> T queryForObject(BaseStringQuery query, Class<T> clazz) {
		Page<T> page = queryForPage(query, clazz);
		Assert.isTrue(page.getTotalElements() < 2, "Expected 1 but found " + page.getTotalElements() + " results");
		return page.getTotalElements() > 0 ? page.getContent().get(0) : null;
	}

	@Override
	public <T> T queryForObject(BaseCriteriaQuery query, Class<T> clazz) {
		Page<T> page = queryForPage(query, clazz);
		Assert.isTrue(page.getTotalElements() < 2, "Expected 1 but found " + page.getTotalElements() + " results");
		return page.getTotalElements() > 0 ? page.getContent().get(0) : null;
	}

	@Override
	public <T> T queryForObject(BaseGetQuery query, Class<T> clazz, BaseResultsMapper mapper) {
		BaseElasticsearchPersistentEntity<T> persistentEntity = getPersistentEntityFor(clazz);
		T entity =null;
		GetRequest request = new GetRequest(persistentEntity.getIndexName(), persistentEntity.getIndexType(), query.getId());
		final GetResponse response = execute(client -> client.get(request));
		if(response.isExists()) {
			entity = mapper.mapResult(response, clazz);
		}
		return entity;
	}

    @Override
    public <T> AggregatedPage<T> queryForPage(BaseSearchQuery query, Class<T> clazz) {
        return queryForPage(query,clazz,resultsMapper);
    }

	@Override
	public <T> Page<T> queryForPage(BaseStringQuery query, Class<T> clazz) {
		return queryForPage(query, clazz, resultsMapper);
	}

	@Override
	public <T> Page<T> queryForPage(BaseStringQuery stringQuery, Class<T> clazz, BaseSearchResultMapper mapper) {
		final SearchRequest request = prepareSearch(stringQuery, clazz);
		request.source().query(wrapperQuery(stringQuery.getSource()));
		SearchResponse response = execute(client -> client.search(request));
		return resultsMapper.mapResults(response, clazz, stringQuery.getPageable());
	}

	@Override
	public <T> Page<T> queryForPage(BaseCriteriaQuery criteriaQuery, Class<T> clazz) {
		QueryBuilder elasticsearchQuery = new BaseCriteriaQueryProcessor().createQueryFromCriteria(criteriaQuery.getCriteria());
		QueryBuilder elasticsearchFilter = new BaseCriteriaFilterProcessor()
				.createFilterFromCriteria(criteriaQuery.getCriteria());
		final SearchRequest request = prepareSearch(criteriaQuery, clazz);

		if (elasticsearchQuery != null) {
			request.source().query(elasticsearchQuery);
		} else {
			request.source().query(QueryBuilders.matchAllQuery());
		}

		if (criteriaQuery.getMinScore() > 0) {
			request.source().minScore(criteriaQuery.getMinScore());
		}
		if (elasticsearchFilter != null)
			request.source().postFilter(elasticsearchFilter);
		SearchResponse response = execute(client -> client.search(request));
		return resultsMapper.mapResults(response, clazz, criteriaQuery.getPageable());
	}

	@Override
    public <T> AggregatedPage<T> queryForPage(BaseSearchQuery query, Class<T> clazz, BaseSearchResultMapper mapper) {
		AggregatedPage<T> result = new AggregatedPageImpl<>();
		long start = System.nanoTime();
		logger.debug("queryForPage start");
		final SearchRequest request = doSearch(prepareSearch(query, clazz), query);
		long requestOk = System.nanoTime();
		logger.debug("hold request ok :{}",(requestOk-start)/1000_000);
		SearchResponse response = execute(client -> client.search(request));
		long responseOk = System.nanoTime();
		logger.debug("hold responseOk ok :{}",(responseOk-requestOk)/1000_000);
		if(RestStatus.OK == response.status()) {
			result = mapper.mapResults(response, clazz, query.getPageable());
		}
		long mapResults = System.nanoTime();
		logger.debug("hold mapResults ok :{}",(mapResults-responseOk)/1000_000);
		return result;
    }

	@Override
	public <T> T query(BaseSearchQuery query, BaseResultsExtractor<T> resultsExtractor) {
		final SearchRequest searchRequest = doSearch(prepareSearch(query), query);
		searchRequest.source().size(1);
		SearchResponse response = execute(client -> client.search(searchRequest));
		return resultsExtractor.extract(response);
	}

	@Override
	public <T> T query(BaseSearchQuery query, Class<T> clazz) {
		final SearchRequest searchRequest = doSearch(prepareSearch(query,clazz), query);
		searchRequest.source().size(1);
		SearchResponse response = execute(client -> client.search(searchRequest));
		return resultsMapper.mapResult(response,clazz);
	}


	@Override
	public <T> List<T> queryForList(BaseSearchQuery query, Class<T> clazz) {
		return queryForPage(query, clazz).getContent();
	}

	@Override
	public <T> List<T> queryForList(BaseStringQuery query, Class<T> clazz) {
		return queryForPage(query,clazz).getContent();
	}

	@Override
	public <T> List<T> queryForList(BaseCriteriaQuery query, Class<T> clazz) {
		return queryForPage(query, clazz).getContent();
	}

	@Override
	public <T> List<String> queryForIds(BaseSearchQuery query) {
		final SearchRequest request = doSearch(prepareSearch(query), query);
		SearchResponse response = execute(client -> client.search(request));
		if(RestStatus.OK == response.status()) {
			logger.debug("searchPage.DSL:{}",request);
			return extractIds(response);
		}
		return extractIds(response);
	}

	@Override
	public <T> CloseableIterator<T> stream(BaseSearchQuery query, Class<T> clazz) {
		return stream(query, clazz, resultsMapper);
	}

	@Override
	public <T> CloseableIterator<T> stream(BaseCriteriaQuery query, Class<T> clazz) {
		final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
		return doStream(scrollTimeInMillis, (BaseScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz), clazz, resultsMapper);
	}

	@Override
	public <T> CloseableIterator<T> stream(BaseSearchQuery query, final Class<T> clazz, final BaseSearchResultMapper mapper) {
		final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
		return doStream(scrollTimeInMillis, (BaseScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz, mapper), clazz, mapper);
	}

	@Deprecated
	private <T> CloseableIterator<T> doStream(final long scrollTimeInMillis, final BaseScrolledPage<T> page, final Class<T> clazz, final BaseSearchResultMapper mapper) {
		return new CloseableIterator<T>() {

			/** As we couldn't retrieve single result with scroll, store current hits. */
			private volatile Iterator<T> currentHits = page.iterator();

			/** The scroll id. */
			private volatile String scrollId = page.getScrollId();

			/** If stream is finished (ie: cluster returns no results. */
			private volatile boolean finished = !currentHits.hasNext();

			@Override
			public void close() {
				try {
					// Clear scroll on cluster only in case of error (cause elasticsearch auto clear scroll when it's done)
					if (!finished && scrollId != null && currentHits != null && currentHits.hasNext()) {
						clearScroll(scrollId);
					}
				} finally {
					currentHits = null;
					scrollId = null;
				}
			}

			@Override
			public boolean hasNext() {
				// Test if stream is finished
				if (finished) {
					return false;
				}
				// Test if it remains hits
				if (currentHits == null || !currentHits.hasNext()) {
					// Do a new request
					final BaseScrolledPage<T> scroll = (BaseScrolledPage<T>) continueScroll(scrollId, scrollTimeInMillis, clazz, mapper);
					// Save hits and scroll id
					currentHits = scroll.iterator();
					finished = !currentHits.hasNext();
					scrollId = scroll.getScrollId();
				}
				return currentHits.hasNext();
			}

			@Override
			public T next() {
				if (hasNext()) {
					return currentHits.next();
				}
				throw new NoSuchElementException();
			}

			@Override
			public void remove() {
				throw new UnsupportedOperationException("remove");
			}
		};
	}


	@Override
	public <T> long count(BaseSearchQuery searchQuery, Class<T> clazz) {
		//todo check
		//filter could not be set into CountRequestBuilder, convert request into search request
		return doCount(prepareSearch(searchQuery, clazz), searchQuery.getQuery(), searchQuery.getFilter());

	}

	@Override
	public <T> long count(BaseCriteriaQuery criteriaQuery, Class<T> clazz) {
		QueryBuilder elasticsearchQuery = new BaseCriteriaQueryProcessor().createQueryFromCriteria(criteriaQuery.getCriteria());
		QueryBuilder elasticsearchFilter = new BaseCriteriaFilterProcessor()
				.createFilterFromCriteria(criteriaQuery.getCriteria());
		// filter could not be set into CountRequestBuilder, convert request into search request
		return doCount(prepareSearch(criteriaQuery, clazz), elasticsearchQuery, elasticsearchFilter);
	}


	@Override
	public <T> long count(BaseSearchQuery query) {
		return count(query, null);
	}

	private long doCount(SearchRequestBuilder countRequestBuilder, QueryBuilder elasticsearchQuery) {

		if (elasticsearchQuery != null) {
			countRequestBuilder.setQuery(elasticsearchQuery);
		}
		return countRequestBuilder.execute().actionGet().getHits().getTotalHits();
	}

	private long doCount(SearchRequest request, QueryBuilder elasticsearchQuery,QueryBuilder elasticsearchFilter) {
		SearchSourceBuilder searchSourceBuilder = request.source();

		if (elasticsearchQuery != null) {
			searchSourceBuilder.query(elasticsearchQuery);
		} else {
			searchSourceBuilder.query(QueryBuilders.matchAllQuery());
		}
		if (elasticsearchFilter != null) {
			searchSourceBuilder.postFilter(elasticsearchFilter);
		}
		searchSourceBuilder.size(0);
		final SearchResponse response = execute(client -> client.search(request));
		return response.getHits().getTotalHits();
	}

	@Override
	public <T> LinkedList<T> multiGet(BaseSearchQuery searchQuery, Class<T> clazz) {
		return resultsMapper.mapResults(getMultiResponse(searchQuery, clazz), clazz);
	}

	private <T> SearchResponse getMultiResponse(BaseQuery searchQuery, Class<T> clazz) {

		String indexName = !isEmpty(searchQuery.getIndices()) ? searchQuery.getIndices().get(0)
				: getPersistentEntityFor(clazz).getIndexName();
		String type = !isEmpty(searchQuery.getTypes()) ? searchQuery.getTypes().get(0)
				: getPersistentEntityFor(clazz).getIndexType();

		Assert.notNull(indexName, "No index defined for Query");
		Assert.notNull(type, "No type define for Query");
		Assert.notEmpty(searchQuery.getIds(), "No Id define for Query");

		SearchSourceBuilder searchSourceBuilder = SearchSourceBuilder.searchSource();

		if (searchQuery.getFields() != null && !searchQuery.getFields().isEmpty()) {
			searchSourceBuilder.fetchSource(new FetchSourceContext(true,toArray(searchQuery.getFields()), null));
		}

		if(!CollectionUtils.isEmpty(searchQuery.getIds())) {
			searchSourceBuilder.postFilter(QueryBuilders.termsQuery("_id",searchQuery.getIds()));
			searchSourceBuilder.size(searchQuery.getIds().size());
		}
		SearchRequest request = new SearchRequest(indexName);
		request.types(type);
		request.source(searchSourceBuilder);
		SearchResponse response = execute(client -> client.search(request));
		if(RestStatus.OK == response.status()) {
			return response;
		}
		return null;
	}

	@Override
	public <T> LinkedList<T> multiGet(BaseSearchQuery searchQuery, Class<T> clazz, BaseMultiGetResultMapper getResultMapper) {
		return getResultMapper.mapResults(getMultiResponse(searchQuery, clazz), clazz);
	}

	@Override
	public String index(BaseIndexQuery query) {
		final UpdateRequest request = putUpdateRequest(query);

		UpdateResponse response = execute(client -> client.update(request));
		String id = response.getId();
		logger.debug("base index reponse status:{},id:{},version:{}",response.status(),id,response.getVersion());
		ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
		if (shardInfo.getTotal() != shardInfo.getSuccessful()) {

		}
		if (shardInfo.getFailed() > 0) {
			final String failureReason = Arrays.stream(shardInfo.getFailures()).map(ReplicationResponse.ShardInfo.Failure::reason).collect(Collectors.joining(";"));
			throw new ElasticsearchException("base index reponse error : "+failureReason);
		}

		// We should call this because we are not going through a mapper.
		if (query.getObject() != null) {
			setPersistentEntityId(query.getObject(), id);
		}
		return id;
	}



	@Override
	public UpdateResponse update(BaseUpdateQuery query) {
		throw new ElasticsearchException("update Deprecated");
	}
	@Override
	public UpdateResponse update(UpdateRequest request) {
		Assert.notNull(request, "No IndexRequest define for Query");
		Assert.notNull(request.index(), "No index defined for Query");
		Assert.notNull(request.type(), "No type define for Query");
		Assert.notNull(request.id(), "No Id define for Query");
		//需要指定id，如果id已存在，以UPSERT的方式更新数据，而不是直接覆盖原有数据。如果false,id 找不到404
		request.docAsUpsert(true);
		BulkRequestSpitUtil.initParamUpdateRequest(request);
		final UpdateResponse response = execute(client -> client.update(request));
		return response;
	}

	@Override
	public <T> UpdateResponse update(UpdateRequest request, Class<T> clazz) {
		Assert.notNull(request, "No IndexRequest define for Query");
		String indexName = getPersistentEntityFor(clazz).getIndexName();
		String type = getPersistentEntityFor(clazz).getIndexType();
		request.index(indexName);
		request.type(type);
		Assert.notNull(request.index(), "No index defined for Query");
		Assert.notNull(request.type(), "No type define for Query");
		Assert.notNull(request.id(), "No Id define for Query");
		//需要指定id，如果id已存在，以UPSERT的方式更新数据，而不是直接覆盖原有数据。如果false,id 找不到404
		request.docAsUpsert(true);
		BulkRequestSpitUtil.initParamUpdateRequest(request);
		final UpdateResponse response = execute(client -> client.update(request));
		return response;
	}

	@Override
	public void bulkIndex(List<BaseIndexQuery> queries) {
		if(!CollectionUtils.isEmpty(queries)) {
			List<BulkRequest> requests = BulkRequestSpitUtil.spitUpdateRequest(() -> queries, p ->putUpdateRequest(p));
			for (BulkRequest item: requests) {
				long start = System.nanoTime();
				final BulkResponse response = execute(client -> client.bulk(item));
				logger.debug("{} - [{}]", Thread.currentThread().getStackTrace()[1].getMethodName(),(System.nanoTime() - start)/1000_000);
				AsyncNotifyOnceListener.checkForBulkUpdateFailure(response);
			}
		}
	}

	@Override
	public void bulkIndexAsync(List<BaseIndexQuery> queries) {
		if(!CollectionUtils.isEmpty(queries)) {
			List<BulkRequest> requests = BulkRequestSpitUtil.spitUpdateRequest(() -> queries, p ->putUpdateRequest(p));
			for (BulkRequest item: requests) {
				long start = System.nanoTime();
				executeAsync(client -> client.bulkAsync(item, new AsyncNotifyOnceListener()));
				logger.debug("{} - [{}]", Thread.currentThread().getStackTrace()[1].getMethodName(),(System.nanoTime() - start)/1000_000);
			}
		}
	}

	@Override
	public void bulkUpdate(List<BaseUpdateQuery> queries) {
		throw new ElasticsearchException("bulkUpdate Deprecated,please use method bulkUpdateRequest!");
	}

	@Override
	public void bulkUpdateRequest(List<UpdateRequest> queries) {
		if(!CollectionUtils.isEmpty(queries)) {
			List<BulkRequest> requests = BulkRequestSpitUtil.spitUpdateRequest(() ->queries, p -> p);
			for (BulkRequest item: requests ) {
				long start = System.nanoTime();
				AsyncNotifyOnceListener.checkForBulkUpdateFailure(execute(client -> client.bulk(item)));
				logger.debug("{} - [{}]", Thread.currentThread().getStackTrace()[1].getMethodName(),(System.nanoTime() - start)/1000_000);
			}
		}
	}

	@Override
	public void bulkUpdateRequestAsync(List<UpdateRequest> queries) {
		if(!CollectionUtils.isEmpty(queries)) {
			List<BulkRequest> requests = BulkRequestSpitUtil.spitUpdateRequest(() ->queries, p -> p);
			for (BulkRequest item: requests ) {
				long start = System.nanoTime();
				executeAsync(client -> client.bulkAsync(item, new AsyncNotifyOnceListener()));
				logger.debug("{} - [{}]", Thread.currentThread().getStackTrace()[1].getMethodName(),(System.nanoTime() - start)/1000_000);
			}
		}
	}

	@Override
	public <T> void bulkUpdateRequest(List<UpdateRequest> queries, Class<T> clazz) {
		if(!CollectionUtils.isEmpty(queries)) {
			String indexName = getPersistentEntityFor(clazz).getIndexName();
			String type = getPersistentEntityFor(clazz).getIndexType();
			for (UpdateRequest query : queries) {
				Assert.notNull(query, "No IndexRequest define for Query");
				query.index(indexName);
				query.type(type);
				BulkRequestSpitUtil.initParamUpdateRequest(query,WriteRequest.RefreshPolicy.NONE);
			}
			bulkUpdateRequest(queries);
		}
	}

	@Override
	public <T> boolean indexExists(Class<T> clazz) {
		return indexExists(getPersistentEntityFor(clazz).getIndexName());
	}

	@Override
	public boolean indexExists(String index) {
		boolean isExist = false;
		IndicesExistsRequest request = new IndicesExistsRequest(index);
		try {
			isExist = client.performRequest(request, RequestConverteUtil::indexExist, BaseRestHighLevelClient::convertIndicesExistsResponse, EMPTY_SET);
			logger.debug("index:[{}],exist:[{}]",index,isExist);
		} catch (IOException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		} catch (ElasticsearchStatusException e) {
			if(RestStatus.NOT_FOUND == e.status()){
				return false;
			}
		}
		return isExist;
	}

	@Override
	public JsonObject typeExists(String index, String type) {
		JsonObject propJson = null;
		TypesExistsRequest request = new TypesExistsRequest(new String[]{index},new String[]{type});
		try {
			JsonObject jsonObject = client.performRequest(request, RequestConverteUtil::typeExist, BaseRestHighLevelClient::convertTypesExistsResponse, EMPTY_SET);
			if(jsonObject.has(index)){
				JsonObject indexJson = jsonObject.getAsJsonObject(index);
				if(indexJson.has("mappings")){
					JsonObject mappingJson = indexJson.getAsJsonObject("mappings");
					if(mappingJson.has(type)){
						propJson = mappingJson.getAsJsonObject(type);
					}
				}
			}
		} catch (ElasticsearchStatusException e) {
			if(RestStatus.NOT_FOUND == e.status()){
				logger.debug("index:[{}],type:[{}] missing ,need mapping",index,type);
				return propJson;
			}
		} catch (IOException e) {
			logger.error("",e);
		}
		return  propJson ;
	}

	@Override
	public <T> boolean deleteIndex(Class<T> clazz) {
//		return deleteIndex(getPersistentEntityFor(clazz).getIndexName());
		return false;
	}

	@Override
	public boolean deleteIndex(String indexName) {
		Assert.notNull(indexName, "No index defined for delete operation");
//		if (indexExists(indexName)) {
//			return client.admin().indices().delete(new DeleteIndexRequest(indexName)).actionGet().isAcknowledged();
//		}
		return false;
	}

	@Override
	public String delete(String indexName, String type, String id) {
		DeleteRequest request = new DeleteRequest(indexName,type,id);
		final DeleteResponse response = execute(client -> client.delete(request));
		return response.getId();
	}

	@Override
	public <T> String delete(Class<T> clazz, String id) {
		BaseElasticsearchPersistentEntity persistentEntity = getPersistentEntityFor(clazz);
		return delete(persistentEntity.getIndexName(), persistentEntity.getIndexType(), id);
	}

	@Override
	public <T> void delete(BaseDeleteQuery deleteQuery, Class<T> clazz) {

		String indexName = Strings.isNotBlank(deleteQuery.getIndex()) ? deleteQuery.getIndex()
				: getPersistentEntityFor(clazz).getIndexName();
		String typeName = Strings.isNotBlank(deleteQuery.getType()) ? deleteQuery.getType()
				: getPersistentEntityFor(clazz).getIndexType();
		Integer pageSize = deleteQuery.getPageSize() != null ? deleteQuery.getPageSize() : 1000;
		Long scrollTimeInMillis = deleteQuery.getScrollTimeInMillis() != null ? deleteQuery.getScrollTimeInMillis()
				: 10000l;

		BaseSearchQuery searchQuery = new BaseNativeSearchQueryBuilder().withQuery(deleteQuery.getQuery()).withIndices(indexName)
				.withTypes(typeName).withPageable(PageRequest.of(0, pageSize)).build();
		BaseSearchResultMapper onlyIdResultMapper = new BaseSearchResultMapper() {
			@Override
			public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {
				List<String> result = new ArrayList<String>();
				for (SearchHit searchHit : response.getHits().getHits()) {
					String id = searchHit.getId();
					result.add(id);
				}
				if (result.size() > 0) {
					return new AggregatedPageImpl<T>((List<T>) result, response.getScrollId());
				}
				return new AggregatedPageImpl<T>(Collections.EMPTY_LIST, response.getScrollId());
			}
		};

		Page<String> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, String.class, onlyIdResultMapper);
//		BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
		List<String> ids = new ArrayList<>();

		do {
			ids.addAll(scrolledResult.getContent());
			scrolledResult = continueScroll(((BaseScrolledPage<T>)scrolledResult).getScrollId(), scrollTimeInMillis, String.class, onlyIdResultMapper);
		} while(scrolledResult.getContent().size() != 0);
		List<BulkRequest> requests = BulkRequestSpitUtil.spitDeleteRequest(ids,indexName,typeName);
		for (BulkRequest item: requests ) {
			execute(client -> client.bulk(item));
		}

		//todo scrollid arrays
		clearScroll(((BaseScrolledPage<T>) scrolledResult).getScrollId());
	}

	@Override
	public <T> void delete(BaseCriteriaQuery criteriaQuery, Class<T> clazz) {
		QueryBuilder elasticsearchQuery = new BaseCriteriaQueryProcessor().createQueryFromCriteria(criteriaQuery.getCriteria());
		Assert.notNull(elasticsearchQuery, "Query can not be null.");
		BaseDeleteQuery deleteQuery = new BaseDeleteQuery();
		deleteQuery.setQuery(elasticsearchQuery);
		delete(deleteQuery, clazz);
	}

	@Override
	public void delete(BaseDeleteQuery deleteQuery) {
		Assert.notNull(deleteQuery.getIndex(), "No index defined for Query");
		Assert.notNull(deleteQuery.getType(), "No type define for Query");
		delete(deleteQuery, null);
	}

	@Override
	public <T> Aggregations aggregation(SearchQuery<T> searchQuery) {
		String index = Strings.isNotEmpty(searchQuery.getIndex()) ? searchQuery.getIndex() :
				getPersistentEntityFor(searchQuery.getClazz()).getIndexName();
		String type = Strings.isNotEmpty(searchQuery.getIndex()) ? searchQuery.getType() :
				getPersistentEntityFor(searchQuery.getClazz()).getIndexType();
		Assert.notNull(index, "No index defined for Query");
		Assert.notNull(type, "No type define for Query");
		SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
		sourceBuilder.query(searchQuery.getBoolQueryBuilder());
		if (searchQuery.getInclude() != null) {
			sourceBuilder.fetchSource(searchQuery.getInclude(), searchQuery.getExclude() != null ? searchQuery.getExclude() : new String[0]);
		}
		if (null!= searchQuery.getAggregation()&& searchQuery.getAggregation().length>0) {
			for (int i = 0; i <searchQuery.getAggregation().length ; i++) {
				sourceBuilder.aggregation(searchQuery.getAggregation()[i]);
			}
		}
		sourceBuilder.size(0);
		SearchRequest request = new SearchRequest(index);
		request.types(type).source(sourceBuilder);
		SearchResponse response = execute(client -> client.search(request));
		return response.getAggregations();
	}


	@Override
	public <T> Response updateByQuery(SearchQuery<T> searchQuery, Script script) {
		String body = String.format("{\n  \"query\": %s,\n  \"script\": {\n    \"inline\": \"%s\",\n    \"lang\": \"painless\"\n  }\n}", searchQuery.getBoolQueryBuilder().toString(), script.getIdOrCode());
		String index = searchQuery.getIndex();
		String type = searchQuery.getType();
		String updatePoint = String.format("/%s/%s/_update_by_query", index, type);
		logger.info("updateByQuery point {}", updatePoint);
		logger.info("updateByQuery body {}", body);
		try {
			Response response = this.restClient.performRequest(HttpPost.METHOD_NAME, updatePoint, Collections.emptyMap(), new StringEntity(body, ContentType.APPLICATION_JSON),
					new Header[]{new BasicHeader("Content-Type", "application/json")});
			logger.info("updateByQuery Response {} {}", response.getStatusLine().getStatusCode(), EntityUtils.toString(response.getEntity()));
			return response;
		} catch (UnsupportedEncodingException var8) {
			logger.error("Exception ", var8);
		} catch (IOException var9) {
			logger.error("Exception ", var9);
		}
		throw new RuntimeException("Response is null");
	}

	@Override
	public <T> List<T> searchList(SearchQuery<T> searchQuery) {
		SearchSourceBuilder sourceBuilder = (new SearchSourceBuilder()).size(searchQuery.getSize()).query(searchQuery.getBoolQueryBuilder());
		if (null != searchQuery.getInclude()) {
			sourceBuilder.fetchSource(searchQuery.getInclude(), searchQuery.getExclude() != null ? searchQuery.getExclude() : new String[0]);
		}
		SearchRequest searchRequest = (new SearchRequest(searchQuery.getIndex()))
				.types(searchQuery.getType())
				.source(sourceBuilder);
		SearchResponse response = execute(client -> client.search(searchRequest));
		return resultsMapper.mapResults(response, searchQuery.getClazz());
	}

	private <T> SearchSourceBuilder prepareScroll(BaseQuery query, Class<T> clazz) {
		setPersistentEntityIndexAndType(query, clazz);
		SearchSourceBuilder searchSourceBuilder = SearchSourceBuilder.searchSource();
		if(query.getPageable().isPaged()){
			searchSourceBuilder.size(query.getPageable().getPageSize());
		}
		if (!isEmpty(query.getFields())) {
			searchSourceBuilder.fetchSource(toArray(query.getFields()), null);
		}
		return searchSourceBuilder;
	}


	private SearchResponse doScroll(SearchSourceBuilder searchSourceBuilder, BaseCriteriaQuery criteriaQuery,long scrollTimeInMillis) {
		Assert.notNull(criteriaQuery.getIndices(), "No index defined for Query");
		Assert.notNull(criteriaQuery.getTypes(), "No type define for Query");
		Assert.notNull(criteriaQuery.getPageable(), "Query.pageable is required for scan & scroll");

		QueryBuilder elasticsearchQuery = new BaseCriteriaQueryProcessor().createQueryFromCriteria(criteriaQuery.getCriteria());
		QueryBuilder elasticsearchFilter = new BaseCriteriaFilterProcessor()
				.createFilterFromCriteria(criteriaQuery.getCriteria());

		if (elasticsearchQuery != null) {
			searchSourceBuilder.query(elasticsearchQuery);
		} else {
			searchSourceBuilder.query(QueryBuilders.matchAllQuery());
		}

		if (elasticsearchFilter != null) {
			searchSourceBuilder.postFilter(elasticsearchFilter);
		}
		Scroll scroll = new Scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
		SearchRequest request = new SearchRequest(criteriaQuery.getIndices().get(0));
		request.types(criteriaQuery.getTypes().get(0))
				.source(searchSourceBuilder);
		request.scroll(scroll);
		SearchResponse response = execute(client -> client.search(request));
		return  response;
	}

	private SearchResponse doScroll(SearchSourceBuilder searchSourceBuilder, BaseSearchQuery searchQuery,long scrollTimeInMillis) {
		Assert.notNull(searchQuery.getIndices(), "No index defined for Query");
		Assert.notNull(searchQuery.getTypes(), "No type define for Query");
		Assert.notNull(searchQuery.getPageable(), "Query.pageable is required for scan & scroll");
		if (searchQuery.getFilter() != null) {
			searchSourceBuilder.postFilter(searchQuery.getFilter());
		}
		if (searchQuery.getQuery() != null) {
			searchSourceBuilder.query(searchQuery.getQuery());
		}
		Scroll scroll = new Scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
		SearchRequest request = new SearchRequest(searchQuery.getIndices().get(0));
		request.types(searchQuery.getTypes().get(0))
				.source(searchSourceBuilder);
		request.scroll(scroll);
		SearchResponse response = execute(client -> client.search(request));
		return  response;
	}

	@Override
	public <T> Page<T> startScroll(long scrollTimeInMillis, BaseSearchQuery searchQuery, Class<T> clazz) {
		SearchResponse response = doScroll(prepareScroll(searchQuery, clazz), searchQuery,scrollTimeInMillis);
		return resultsMapper.mapResults(response, clazz, null);
	}


	@Override
	public <T> Page<T> startScroll(long scrollTimeInMillis, BaseSearchQuery searchQuery, Class<T> clazz, BaseSearchResultMapper mapper) {
		SearchResponse response = doScroll(prepareScroll(searchQuery, clazz), searchQuery,scrollTimeInMillis);
		return mapper.mapResults(response, clazz, null);
	}

	@Override
	public <T> Page<T> startScroll(long scrollTimeInMillis, BaseCriteriaQuery criteriaQuery, Class<T> clazz) {
		SearchResponse response = doScroll(prepareScroll(criteriaQuery, clazz), criteriaQuery,scrollTimeInMillis);
		return resultsMapper.mapResults(response, clazz, null);
	}


	@Override
	public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz) {
		return this.continueScroll(scrollId,scrollTimeInMillis,clazz,this.resultsMapper);
	}

	@Override
	public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz, BaseSearchResultMapper mapper) {
		SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
		scrollRequest.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
		final SearchResponse response = execute(client -> client.searchScroll(scrollRequest));
		return mapper.mapResults(response, clazz, Pageable.unpaged());
	}

	@Override
	public void clearScroll(String scrollId) {
		ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
		clearScrollRequest.setScrollIds(Lists.newArrayList(scrollId));
		executeAsync(client -> client.clearScrollAsync(clearScrollRequest, new ActionListener() {
			@Override
			public void onResponse(Object o) {
				logger.info("queryContactByModeScroll clear scroll result:{}", o);
			}
			@Override
			public void onFailure(Exception e) {
				logger.info("queryContactByModeScroll clear scroll fail:{}", Throwables.getStackTraceAsString(e));
			}
		}));
	}



	@Override
	public void clearScroll(List<String> scrollId){
		ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
		clearScrollRequest.setScrollIds(scrollId.stream().distinct().collect(Collectors.toList()));
		executeAsync(client -> client.clearScrollAsync(clearScrollRequest, new ActionListener() {
			@Override
			public void onResponse(Object o) {
				logger.info("queryContactByModeScroll clear scroll result:{}", o);
			}
			@Override
			public void onFailure(Exception e) {
				logger.info("queryContactByModeScroll clear scroll fail:{}", Throwables.getStackTraceAsString(e));
			}
		}));
	}


	private SearchRequest doSearch(SearchRequest searchRequest, BaseSearchQuery searchQuery) {
		SearchSourceBuilder searchSourceBuilder =searchRequest.source();

		if (searchQuery.getFilter() != null) {
			searchSourceBuilder.postFilter(searchQuery.getFilter());
		}

		if (!isEmpty(searchQuery.getElasticsearchSorts())) {
			for (SortBuilder sort : searchQuery.getElasticsearchSorts()) {
				searchSourceBuilder.sort(sort);
			}
		}

		if (!searchQuery.getScriptFields().isEmpty()) {
			//_source should be return all the time
			//searchRequest.addStoredField("_source");
			for (SearchSourceBuilder.ScriptField scriptedField : searchQuery.getScriptFields()) {
				searchSourceBuilder.scriptField(scriptedField.fieldName(), scriptedField.script());
			}
		}

		if (searchQuery.getHighlightFields() != null) {
			for (HighlightBuilder.Field highlightField : searchQuery.getHighlightFields()) {
				searchSourceBuilder.highlighter(new HighlightBuilder().field(highlightField));
			}
		}

		if (!isEmpty(searchQuery.getIndicesBoost())) {
//			for (IndexBoost indexBoost : searchQuery.getIndicesBoost()) {
//				searchRequest.addIndexBoost(indexBoost.getIndexName(), indexBoost.getBoost());
//			}
		}

		if (!isEmpty(searchQuery.getAggregations())) {
			for (AbstractAggregationBuilder aggregationBuilder : searchQuery.getAggregations()) {
				searchSourceBuilder.aggregation(aggregationBuilder);
			}
		}
		searchSourceBuilder.query(searchQuery.getQuery());
		searchRequest.source(searchSourceBuilder);

		return searchRequest;
//		return getSearchResponse().execute());
	}

//	@Deprecated
//	private SearchResponse getSearchResponse(ListenableActionFuture<SearchResponse> response) {
//		return searchTimeout == null ? response.actionGet() : response.actionGet(searchTimeout);
//	}

	private <T> boolean createIndexIfNotCreated(Class<T> clazz) {
		return indexExists(getPersistentEntityFor(clazz).getIndexName()) || createIndexWithSettings(clazz);
	}

	private <T> boolean createIndexWithSettings(Class<T> clazz) {
		if (clazz.isAnnotationPresent(Setting.class)) {
			String settingPath = clazz.getAnnotation(Setting.class).settingPath();
			if (Strings.isNotBlank(settingPath)) {
				String settings = readFileFromClasspath(settingPath);
				if (Strings.isNotBlank(settings)) {
					return createIndex(getPersistentEntityFor(clazz).getIndexName(), settings);
				}
			} else {
				logger.info("settingPath in @Setting has to be defined. Using default instead.");
			}
		}
		return createIndex(getPersistentEntityFor(clazz).getIndexName(), getDefaultSettings(getPersistentEntityFor(clazz)));
	}

	@Override
	public boolean createIndex(String indexName, Object settings) {
		boolean response = false;
		final CreateIndexRequest indexRequest = Requests.createIndexRequest(indexName);
		if (settings instanceof String) {
			indexRequest.settings(String.valueOf(settings));
		} else if (settings instanceof Map) {
			indexRequest.settings((Map) settings);
		} else if (settings instanceof XContentBuilder) {
			indexRequest.settings((XContentBuilder) settings);
		}
		try{
			response = client.performRequest(indexRequest, RequestConverteUtil::index, BaseRestHighLevelClient::convertCreateIndexResponse, emptySet());
			logger.debug("index:[{}],create:[{}]",indexName,response);
		} catch (IOException | ElasticsearchStatusException e) {
			logger.error("createIndex",e);
			throw new com.base.data.elasticsearch.ElasticsearchException(e.getMessage());
		}
		return response;
	}

	@Override
	public <T> boolean createIndex(Class<T> clazz, Object settings) {
		return createIndex(getPersistentEntityFor(clazz).getIndexName(), settings);
	}



	@Override
	public <T> boolean updateTypeMapping(String index, String type, Class<T> tClass, JsonObject existPropJson) {
		Assert.notNull(index, "No index defined for getSettings");
		Assert.notNull(type, "No index defined for getSettings");
		Boolean response = false;
		try {
			JsonObject properties = new JsonObject();
			if(existPropJson.has(FIELD_PROPERTIES)){
				properties =existPropJson.getAsJsonObject(FIELD_PROPERTIES);
			}
			BaseElasticsearchPersistentEntity<T> persistentEntity = getPersistentEntityFor(tClass);

			BaseElasticsearchPersistentProperty property = persistentEntity.getRequiredIdProperty();

			XContentBuilder mapping = BaseMappingBuilder.buildMappingDiff(tClass, persistentEntity.getIndexType(),
						property.getFieldName(), persistentEntity.getParentType(),properties);
			if(null!= mapping) {
				logger.debug("index:[{}],type:[{}],update-mapping:[{}]",index,type,mapping.string());
				BaseCreateIndexRequest request = new BaseCreateIndexRequest();
				request.index(index).type(type).source(mapping).create(false);
				response = client.performRequest(request, RequestConverteUtil::index, BaseRestHighLevelClient::convertCreateIndexResponse, emptySet());
				logger.debug("index:[{}],type:[{}],update-mapping:[{}]",index,type,response);
			}else {
				logger.debug("index:[{}],type:[{}] exist,check no need sync ...",index,type);
			}
		} catch (IOException | ElasticsearchStatusException e) {
			logger.error("updateTypeMapping",e);
			throw new com.base.data.elasticsearch.ElasticsearchException(e.getMessage());
		}
		return response;
	}

	@Override
	public <T> boolean updateTypeMapping(Class<T> tClass, JsonObject existPropJson) {
		BaseElasticsearchPersistentEntity<T> persistentEntity = getPersistentEntityFor(tClass);
		final String indexName = persistentEntity.getIndexName();
		final String indexType = persistentEntity.getIndexType();

		return updateTypeMapping(indexName,indexType,tClass,existPropJson);
	}

	private <T> Map getDefaultSettings(BaseElasticsearchPersistentEntity<T> persistentEntity) {

		if (persistentEntity.isUseServerConfiguration())
			return new HashMap();

		return new MapBuilder<String, String>().put("index.number_of_shards", String.valueOf(persistentEntity.getShards()))
				.put("index.number_of_replicas", String.valueOf(persistentEntity.getReplicas()))
				.put("index.refresh_interval", persistentEntity.getRefreshInterval())
				.put("index.store.type", persistentEntity.getIndexStoreType()).map();
	}

	@Override
	public <T> Map getSetting(Class<T> clazz) {
		return getSetting(getPersistentEntityFor(clazz).getIndexName());
	}

	@Override
	public Map getSetting(String indexName) {
		Assert.notNull(indexName, "No index defined for getSettings");
//		return client.admin().indices().getSettings(new GetSettingsRequest()).actionGet().getIndexToSettings()
//				.get(indexName).getAsMap();
		return null;
	}

	private <T> SearchRequest prepareSearch(BaseQuery query, Class<T> clazz) {
		setPersistentEntityIndexAndType(query, clazz);
		return prepareSearch(query);
	}

	private SearchRequest prepareSearch(BaseQuery query) {
		Assert.notNull(query.getIndices(), "No index defined for Query");
		Assert.notEmpty(query.getIndices(), "Empty index defined for Query");
		Assert.notNull(query.getTypes(), "No type defined for Query");
		Assert.notEmpty(query.getTypes(), "Empty type defined for Query");

		SearchRequest searchRequest = new SearchRequest(toArray(query.getIndices()));
		searchRequest.types(toArray(query.getTypes()));
		searchRequest.searchType(query.getSearchType());
		int startRecord = 0;
		SearchSourceBuilder searchRequestBuilder = SearchSourceBuilder.searchSource();

		if (query.getSourceFilter() != null) {
			BaseSourceFilter sourceFilter = query.getSourceFilter();
			searchRequestBuilder.fetchSource(sourceFilter.getIncludes(), sourceFilter.getExcludes());
		}

		if (query.getPageable().isPaged()) {
			final int pageSize = query.getPageable().getPageSize();
			startRecord = query.getPageable().getPageNumber() * pageSize;
			if(pageSize >= 10_000) {
				logger.info("prepareSearch over 10_000, size:{}",pageSize);
			}
			searchRequestBuilder.size(pageSize);
		}
		searchRequestBuilder.from(startRecord);

		if (!query.getFields().isEmpty()) {
			searchRequestBuilder.fetchSource(toArray(query.getFields()),null);
		}

		if (query.getSort() != null) {
			for (Sort.Order order : query.getSort()) {
				searchRequestBuilder.sort(order.getProperty(),
						order.getDirection() == Sort.Direction.DESC ? SortOrder.DESC : SortOrder.ASC);
			}
		}

		if (query.getMinScore() > 0) {
			searchRequestBuilder.minScore(query.getMinScore());
		}
		searchRequest.source(searchRequestBuilder);
		return searchRequest;
	}

	private  UpdateRequest putUpdateRequest (BaseIndexQuery query ) {
		final IndexRequest indexRequest = prepareIndex(query);
		UpdateRequest request = new UpdateRequest(indexRequest.index(), indexRequest.type(), indexRequest.id())
				.doc(indexRequest).docAsUpsert(true);
		BulkRequestSpitUtil.initParamUpdateRequest(request,WriteRequest.RefreshPolicy.NONE);
		return request;
	}

	private IndexRequest prepareIndex(BaseIndexQuery query) {
		try {
			String indexName = Strings.isBlank(query.getIndexName())
					? retrieveIndexNameFromPersistentEntity(query.getObject().getClass())[0] : query.getIndexName();
			String type = Strings.isBlank(query.getType()) ? retrieveTypeFromPersistentEntity(query.getObject().getClass())[0]
					: query.getType();
			IndexRequest indexRequest = new IndexRequest(indexName,type);
			final Object entity = query.getObject();

			if (entity != null) {
				String entityId = RefectUtil.refectEntityIdValue(entity);
				String jsonParam = JacksonUtil.toString(entity);
				indexRequest.id(entityId).source(jsonParam, XContentType.JSON);
			}

			if (query.getVersion() != null) {
				indexRequest.version(query.getVersion());
				indexRequest.versionType(EXTERNAL);
			}

			if (query.getParentId() != null) {
				indexRequest.parent(query.getParentId());
			}

			return indexRequest;
		} catch (Exception e) {
			throw new ElasticsearchException("failed to index the document [id: " + query.getId() + "]", e);
		}
	}



	@Override
	public BaseElasticsearchPersistentEntity getPersistentEntityFor(Class clazz) {
		Assert.isTrue(clazz.isAnnotationPresent(Document.class), "Unable to identify index name. " + clazz.getSimpleName()
				+ " is not a Document. Make sure the document class is annotated with @Document(indexName=\"foo\")");
		return elasticsearchConverter.getMappingContext().getRequiredPersistentEntity(clazz);
	}

	private String getPersistentEntityId(Object entity) {

		BaseElasticsearchPersistentEntity<?> persistentEntity = getPersistentEntityFor(entity.getClass());
		Object identifier = persistentEntity.getIdentifierAccessor(entity).getIdentifier();

		if (identifier != null){
			return identifier.toString();
		}

		return null;
	}

	private void setPersistentEntityId(Object entity, String id) {

		BaseElasticsearchPersistentEntity<?> persistentEntity = getPersistentEntityFor(entity.getClass());
		BaseElasticsearchPersistentProperty idProperty = persistentEntity.getIdProperty();

		// Only deal with text because ES generated Ids are strings !

		if (idProperty != null && idProperty.getType().isAssignableFrom(String.class)) {
			persistentEntity.getPropertyAccessor(entity).setProperty(idProperty, id);
		}
	}

	private void setPersistentEntityIndexAndType(BaseQuery query, Class clazz) {
		if (query.getIndices().isEmpty()) {
			query.addIndices(retrieveIndexNameFromPersistentEntity(clazz));
		}
		if (query.getTypes().isEmpty()) {
			query.addTypes(retrieveTypeFromPersistentEntity(clazz));
		}
	}

	private String[] retrieveIndexNameFromPersistentEntity(Class clazz) {
		if (clazz != null) {
			return new String[] { getPersistentEntityFor(clazz).getIndexName() };
		}
		return null;
	}

	private String[] retrieveTypeFromPersistentEntity(Class clazz) {
		if (clazz != null) {
			return new String[] { getPersistentEntityFor(clazz).getIndexType() };
		}
		return null;
	}

	private List<String> extractIds(SearchResponse response) {
		List<String> ids = new ArrayList<>();
		for (SearchHit hit : response.getHits()) {
			if (hit != null) {
				ids.add(hit.getId());
			}
		}
		return ids;
	}

	@Override
	public void setApplicationContext(ApplicationContext context) throws BeansException {
		if (elasticsearchConverter instanceof ApplicationContextAware) {
			((ApplicationContextAware) elasticsearchConverter).setApplicationContext(context);
		}
	}

	private static String[] toArray(List<String> values) {
		String[] valuesAsArray = new String[values.size()];
		return values.toArray(valuesAsArray);
	}



	protected BaseResultsMapper getResultsMapper() {
		return resultsMapper;
	}

	public static String readFileFromClasspath(String url) {
		StringBuilder stringBuilder = new StringBuilder();
		BufferedReader bufferedReader = null;

		try {
			ClassPathResource classPathResource = new ClassPathResource(url);
			InputStreamReader inputStreamReader = new InputStreamReader(classPathResource.getInputStream());
			bufferedReader = new BufferedReader(inputStreamReader);
			String line;

			String lineSeparator = System.getProperty("line.separator");
			while ((line = bufferedReader.readLine()) != null) {
				stringBuilder.append(line).append(lineSeparator);
			}
		} catch (Exception e) {
			logger.debug(String.format("Failed to load file from url: %s: %s", url, e.getMessage()));
			return null;
		} finally {
			if (bufferedReader != null)
				try {
					bufferedReader.close();
				} catch (IOException e) {
					logger.debug(String.format("Unable to close buffered reader.. %s", e.getMessage()));
				}
		}

		return stringBuilder.toString();
	}

	public SearchResponse suggest(SuggestBuilder suggestion, String... indices) {
//		return client.prepareSearch(indices).suggest(suggestion).get();
		return  null;
	}

	public SearchResponse suggest(SuggestBuilder suggestion, Class clazz) {
		return suggest(suggestion, retrieveIndexNameFromPersistentEntity(clazz));
	}
}
